package com.lianjia.sh.se.kafka.client;

import com.lianjia.sh.se.kafka.client.config.ClientConsumerDefaultConfig;
import com.lianjia.sh.se.kafka.client.config.CustomDefaultListenerConfig;
import com.lianjia.sh.se.kafka.client.constants.ConsumerCons;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * Created by yangsp1 on 16/7/28.
 */
@Component
public class KafkaConsumerFactory {
    @Autowired
    ClientConsumerDefaultConfig consumerDefaultConfig;

    @Autowired
    CustomDefaultListenerConfig customDefaultListenerConfig;

    //缓存
    Map<Object,Consumer> consumerCache = new HashMap<>();

    /**
     * 创建：消费者
     * @return
     */
    public Consumer createConsumer(Object bean, Properties configs){
        if (this.consumerCache.get(bean) == null) {
            synchronized (bean) {
                if (this.consumerCache.get(bean) == null) {
                    Properties populsteConfigs = populateConfigs(configs);

                    //生成key和value反序列化对象
                    String keyDeserializerClass = (String)populsteConfigs.get(ConsumerCons.KEY_DESERIALIZER_NAME);
                    String valueDeserializerClass = (String)populsteConfigs.get(ConsumerCons.VALUE_DESERIALIZER_NAME);

                    Deserializer keyDeserializer = null;
                    try {
                        keyDeserializer = (Deserializer) Class.forName(keyDeserializerClass).newInstance();
                    }catch (Exception e){
                        keyDeserializer = new StringDeserializer();
                    }

                    Deserializer valueDeserializer = null;
                    try {
                        valueDeserializer = (Deserializer) Class.forName(valueDeserializerClass).newInstance();
                    }catch (Exception e){
                        valueDeserializer = new StringDeserializer();
                    }

                    //生成消费者并缓存
                    this.consumerCache.put(bean,new KafkaConsumer(populsteConfigs, keyDeserializer, valueDeserializer));
                }
            }
        }
        return this.consumerCache.get(bean);
    }


    /**
     * 计算消费都配置
     * @param customConfigs
     * @return
     */
    private Properties populateConfigs(Properties customConfigs){
        //首先使用listener自定义配置
        Properties config = populateCustomConfigs(customConfigs);

        //其次使用应用统一自定义默认配置
        config = populateAppDefaultCustomConfigs(config);

        //全局统一默认配置
        config = populateCommonDefaultConfigs(config);

        //最后：使用kafak原始的默认值
        return config;
    }


    /**
     * listener自身的自定义配置
     * @return
     */
    private Properties populateCustomConfigs(Properties customConfigs){
        Properties rs = new Properties();
        rs.putAll(customConfigs);
        return rs;
    }

    /**
     * 默认的自定义配置（应用级)
     * @return
     */
    private Properties populateAppDefaultCustomConfigs(Properties configs){
        Properties rs = new Properties();
        rs.putAll(configs);
        //broker 地址列表
        String bootstrapServers = (String)configs.get(ConsumerCons.BOOTSTRAP_SERVERS_NAME);
        String bootstrapServersCustom = customDefaultListenerConfig.getBootstrapServers();
        if(StringUtils.isBlank(bootstrapServers) && StringUtils.isNotBlank(bootstrapServersCustom)){
            rs.put(ConsumerCons.BOOTSTRAP_SERVERS_NAME,bootstrapServersCustom);
        }

        //group.id
        String groupId = (String)configs.get(ConsumerCons.GROUP_ID_NAME);
        String groupIdCustom = customDefaultListenerConfig.getGroupId();
        if(StringUtils.isBlank(groupId) && StringUtils.isNotBlank(groupIdCustom)){
            rs.put(ConsumerCons.GROUP_ID_NAME,groupIdCustom);
        }

        //auto.commit.interval.ms
        Long autoCommitIntervalMs = (Long) configs.get(ConsumerCons.AUTO_COMMIT_INTERVAL_MS_NAME);
        Long autoCommitIntervalMsCustom = customDefaultListenerConfig.getAutoCommitIntervalMs();
        if((autoCommitIntervalMs == null || autoCommitIntervalMs <= 0) && (autoCommitIntervalMsCustom != null && autoCommitIntervalMsCustom > 0)){
            rs.put(ConsumerCons.AUTO_COMMIT_INTERVAL_MS_NAME,autoCommitIntervalMsCustom);
        }

        //key.deserializer
        String keyDeserializer = (String) configs.get(ConsumerCons.KEY_DESERIALIZER_NAME);
        String keyDeserializerCustom = customDefaultListenerConfig.getKeyDeserializer();
        if(StringUtils.isBlank(keyDeserializer) && StringUtils.isNotBlank(keyDeserializerCustom)){
            rs.put(ConsumerCons.KEY_DESERIALIZER_NAME,keyDeserializerCustom);
        }

        //value.deserializer
        String valueDeserializer = (String) configs.get(ConsumerCons.VALUE_DESERIALIZER_NAME);
        String valueDeserializerCustom = customDefaultListenerConfig.getValueDeserializer();
        if(StringUtils.isBlank(valueDeserializer) && StringUtils.isNotBlank(valueDeserializerCustom)){
            rs.put(ConsumerCons.VALUE_DESERIALIZER_NAME,valueDeserializerCustom);
        }

        //client.id
//        String clientId = (String) configs.get(ConsumerCons.CLIENT_ID_NAME);
//        String clientIdCustom = customDefaultListenerConfig.getClientId();
//        if(StringUtils.isBlank(clientId) && StringUtils.isNotBlank(clientIdCustom)){
//            rs.put(ConsumerCons.CLIENT_ID_NAME,clientIdCustom);
//        }

        //fetch.min.bytes
        Integer fetchMinBytes = (Integer) configs.get(ConsumerCons.FETCH_MIN_BYTES_NAME);
        Integer fetchMinBytesCustom = customDefaultListenerConfig.getFetchＭinBytes();
        if((fetchMinBytes == null || fetchMinBytes <= 0) && (fetchMinBytesCustom != null && fetchMinBytesCustom > 0)){
            rs.put(ConsumerCons.FETCH_MIN_BYTES_NAME,fetchMinBytesCustom);
        }

        //max.poll.records
        Long maxPollRecords = (Long) configs.get(ConsumerCons.MAX_POLL_RECORDS_NAME);
        Integer maxPollRecordsCustom = customDefaultListenerConfig.getMaxPollRecords();
        if((maxPollRecords == null || maxPollRecords <= 0) && (maxPollRecordsCustom != null && maxPollRecordsCustom > 0)){
            rs.put(ConsumerCons.MAX_POLL_RECORDS_NAME,maxPollRecordsCustom);
        }

        //request.timeout.ms
        Integer requestTimeoutMs = (Integer) configs.get(ConsumerCons.REQUEST_TIMEOUT_MS_NAME);
        Integer requestTimeoutMsCustom = customDefaultListenerConfig.getRequestTimeoutMs();
        if((requestTimeoutMs == null || requestTimeoutMs <= 0) && (requestTimeoutMsCustom != null && requestTimeoutMsCustom > 0)){
            rs.put(ConsumerCons.REQUEST_TIMEOUT_MS_NAME,requestTimeoutMsCustom);
        }

        return rs;
    }

    /**
     * 全局统一默认配置
     * @return
     */
    private Properties populateCommonDefaultConfigs(Properties configs){
        Properties rs = new Properties();
        rs.putAll(configs);
        String bootstrapServers = (String)configs.get(ConsumerCons.BOOTSTRAP_SERVERS_NAME);
        if(StringUtils.isBlank(bootstrapServers) && StringUtils.isNotBlank(consumerDefaultConfig.getBootstrapServers())){
            rs.put(ConsumerCons.BOOTSTRAP_SERVERS_NAME, consumerDefaultConfig.getBootstrapServers());
        }

        //group.id
        String groupId = (String)configs.get(ConsumerCons.GROUP_ID_NAME);
        if(StringUtils.isBlank(groupId) && StringUtils.isNotBlank(consumerDefaultConfig.getGroupId())){
            rs.put(ConsumerCons.GROUP_ID_NAME,consumerDefaultConfig.getGroupId());
        }

        //auto.commit.interval.ms
        Long autoCommitIntervalMs = (Long) configs.get(ConsumerCons.AUTO_COMMIT_INTERVAL_MS_NAME);
        if((autoCommitIntervalMs == null || autoCommitIntervalMs <= 0)
                && (consumerDefaultConfig.getAutoCommitIntervalMs() != null && consumerDefaultConfig.getAutoCommitIntervalMs() > 0)){
            rs.put(ConsumerCons.AUTO_COMMIT_INTERVAL_MS_NAME,consumerDefaultConfig.getAutoCommitIntervalMs());
        }


        //key.deserializer
        String keyDeserializer = (String) configs.get(ConsumerCons.KEY_DESERIALIZER_NAME);
        if(StringUtils.isBlank(keyDeserializer) && StringUtils.isNotBlank(consumerDefaultConfig.getKeyDeserializer())){
            rs.put(ConsumerCons.KEY_DESERIALIZER_NAME,consumerDefaultConfig.getKeyDeserializer());
        }

        //value.deserializer
        String valueDeserializer = (String) configs.get(ConsumerCons.VALUE_DESERIALIZER_NAME);
        if(StringUtils.isBlank(valueDeserializer) && StringUtils.isNotBlank(consumerDefaultConfig.getValueDeserializer())){
            rs.put(ConsumerCons.VALUE_DESERIALIZER_NAME,consumerDefaultConfig.getValueDeserializer());
        }

        //client.id
//        String clientId = (String) configs.get(ConsumerCons.CLIENT_ID_NAME);
//        if(StringUtils.isBlank(clientId) && StringUtils.isNotBlank(consumerDefaultConfig.getClientId())){
//            rs.put(ConsumerCons.CLIENT_ID_NAME,consumerDefaultConfig.getClientId());
//        }

        //fetch.min.bytes
        Long fetchMinBytes = (Long) configs.get(ConsumerCons.FETCH_MIN_BYTES_NAME);
        if((fetchMinBytes == null || fetchMinBytes <= 0)
                && ( consumerDefaultConfig.getFetchMinBytes() != null && consumerDefaultConfig.getFetchMinBytes() > 0)){
            rs.put(ConsumerCons.FETCH_MIN_BYTES_NAME,consumerDefaultConfig.getFetchMinBytes());
        }

        //max.poll.records
        Long maxPollRecords = (Long) configs.get(ConsumerCons.MAX_POLL_RECORDS_NAME);
        if((maxPollRecords == null || maxPollRecords <= 0)
                && ( consumerDefaultConfig.getMaxPollRecords() != null && consumerDefaultConfig.getMaxPollRecords() > 0)){
            rs.put(ConsumerCons.MAX_POLL_RECORDS_NAME,consumerDefaultConfig.getMaxPollRecords());
        }

        //request.timeout.ms
        Long requestTimeoutMs = (Long) configs.get(ConsumerCons.REQUEST_TIMEOUT_MS_NAME);
        if((requestTimeoutMs == null || requestTimeoutMs <= 0)
                && (consumerDefaultConfig.getRequestTimeoutMs() != null && consumerDefaultConfig.getRequestTimeoutMs() > 0)){
            rs.put(ConsumerCons.REQUEST_TIMEOUT_MS_NAME,consumerDefaultConfig.getRequestTimeoutMs());
        }
        return rs;
    }
}
